package tools

import (
	"context"
	"encoding/json"
	"errors"
	"time"

	"github.com/segmentio/kafka-go"
)

/*
 +----------------------------------------------------------------------
 + Title        : kafka
 + Author       : huwl
 + Version      : 1.0.0
 + Initial-Time : 2021-01-06
 + Last-time    : 这个文件最后修改的时间 + 修改人的名称
 + Desc         : olivere插件版elasticsearch
 +----------------------------------------------------------------------
*/

type SegmentioKafka struct {
	kafkaClient *kafka.Conn
}

type KafkaConf struct {
	Address string `yaml:"address"`
}

/**
 * 创建 SegmentioKafka
 *
 * @param string network ---------------------------------------------------必传,网络【tcp】
 * @param string topic -----------------------------------------------------必传
 * @param int partition ----------------------------------------------------必传
 * @return *Kafka, error
 * @author huwl
 */
func NewSegmentioKafka(network, topic string, partition int) (*SegmentioKafka, error) {
	// 读取本地 kafka 文件配置数据
	kafkaConfig, err := GetConfig("kafka", "yml")
	if err != nil {
		return nil, err
	}

	kafkaConf := &KafkaConf{
		Address: kafkaConfig.GetString(Env + ".address"),
	}

	conn, err := kafka.DialLeader(context.Background(), network, kafkaConf.Address, topic, partition)
	if err != nil {
		return nil, err
	}

	return &SegmentioKafka{
		kafkaClient: conn,
	}, nil
}

/**
 * 生产者
 *
 * @param string topic -----------------------------------------------------必传,主题
 * @param map[string]interface{} message -----------------------------------必传,消息
 * @return error
 * @author huwl
 */
func (s *SegmentioKafka) Producer(topic string, message map[string]interface{}) error {
	if len(message) <= 0 {
		return errors.New("消息不能为空")
	}

	s.kafkaClient.SetWriteDeadline(time.Now().Add(10 * time.Second))
	m, err := json.Marshal(message)
	if err != nil {
		return err
	}
	_, err = s.kafkaClient.WriteMessages(kafka.Message{
		Topic: topic,
		Value: m,
	})
	if err != nil {
		return err
	}

	return nil
}
